package com.zheng.cms.rocketmq.listener;


import com.alibaba.rocketmq.client.consumer.listener.*;
import com.alibaba.rocketmq.common.message.MessageExt;
import org.apache.log4j.Logger;


import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

/**
 *
 //设置一个Listener，主要进行消息的逻辑处理
 //注意这里使用的是MessageListenerOrderly这个接口
 //这里设置的是一个consumer的消费策略
 //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
 //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
 //CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
 consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

 作者：冯先生的笔记
 链接：https://www.jianshu.com/p/11e875074a8f
 來源：简书
 简书著作权归作者所有，任何形式的转载都请联系作者获得授权并注明出处。


 * 实现了MessageListenerOrderly表示一个队列只会被一个线程取到，第二个线程无法访问这个队列
 *

 * */
public class OrderMessageListener implements MessageListenerOrderly {

    private Logger log = Logger.getLogger(getClass());

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
//        AtomicLong consumeTimes = new AtomicLong(0);
//        context.setAutoCommit(true); // 设置自动提交
        for (MessageExt msg : msgs) {
                log.info("OrderMessageListener,msg : " + new String(msg.getBody())+" key:"+msg.getKeys());
            }
            return ConsumeOrderlyStatus.SUCCESS;


    }
}
